import pandas as pd  # used to construct DataFrame instances
# imports
import pytest  # used for our unit tests
from langflow.api.utils.kb_helpers import calculate_text_metrics


def test_empty_dataframe_returns_zero():
    # An entirely empty DataFrame with no columns should produce zero counts.
    df = pd.DataFrame()  # real DataFrame instance with no columns and no rows
    total_words, total_chars = calculate_text_metrics(df, ["any_column"])


def test_basic_single_column_simple():
    # Basic functionality: single text column with typical strings.
    df = pd.DataFrame(
        {
            "text": [
                "hello world",  # 2 words, 11 characters (includes the space)
                "test",         # 1 word, 4 characters
                "",             # 0 words, 0 characters
            ]
        }
    )
    # Only the 'text' column is measured
    words, chars = calculate_text_metrics(df, ["text"])


def test_multiple_columns_and_missing_column():
    # Verify function handles multiple text columns and skips missing ones.
    df = pd.DataFrame(
        {
            "a": ["x y", "z"],  # 'x y' -> 2 words, 3 chars; 'z' -> 1 word, 1 char
            "b": [100, 200],    # numeric values will be astyped to '100', '200'
        }
    )
    # Include a non-existent column name 'missing' which should be ignored.
    words, chars = calculate_text_metrics(df, ["a", "missing", "b"])


def test_text_columns_empty_list_returns_zero_even_with_data():
    # If text_columns is empty, no columns are processed even if DataFrame has data.
    df = pd.DataFrame({"col": ["a b c", "d e"]})
    words, chars = calculate_text_metrics(df, [])


def test_nan_and_none_behavior_counts_as_strings():
    # This test documents the actual behavior: the function astypes to str before fillna,
    # which means None and NaN become the literal strings 'None' and 'nan' respectively,
    # and are therefore counted as words and characters.
    df = pd.DataFrame(
        {
            "c": [
                None,            # astype(str) -> 'None' => 1 word, 4 chars
                float("nan"),    # astype(str) -> 'nan'  => 1 word, 3 chars
                "",              # '' -> 0 words, 0 chars
                "nan",           # literal 'nan' -> 1 word, 3 chars
                "None",          # literal 'None' -> 1 word, 4 chars
            ]
        }
    )
    words, chars = calculate_text_metrics(df, ["c"])


def test_duplicate_columns_are_counted_multiple_times():
    # If the same column name appears more than once in text_columns,
    # the implementation iterates and will double-count that column.
    df = pd.DataFrame({"t": ["a b", "c"]})  # words 2 + 1 = 3, chars 3 + 1 = 4
    # Provide the same column twice; expected result should be doubled.
    words, chars = calculate_text_metrics(df, ["t", "t"])


def test_non_string_types_are_converted_to_str_and_counted():
    # Ensure different non-string types are converted to their string representations
    # and then counted for words and characters as the function does via astype(str).
    mixed = [123, 45.6, True, ["list"], {"k": "v"}]
    df = pd.DataFrame({"mixed": mixed})
    words, chars = calculate_text_metrics(df, ["mixed"])
    # Compute expected by applying Python's str conversion to each element, mirroring astype(str)
    expected_words = sum(len(str(x).split()) for x in mixed)
    expected_chars = sum(len(str(x)) for x in mixed)


def test_large_scale_correctness_with_1000_rows():
    # Large-scale test: construct 1000 rows and multiple columns to ensure scalability
    n = 1000
    col1 = [("a" * (i % 10)) for i in range(n)]  # varying lengths 0..9, each non-empty -> 1 word
    # create repeated 'b' tokens with spaces, stripped so split() yields exact counts
    col2 = [(" ".join(["b"] * (i % 5))) for i in range(n)]
    # create a column with up to 6 tokens
    col3 = [(" ".join(["x"] * (i % 7))) for i in range(n)]

    df = pd.DataFrame({"c1": col1, "c2": col2, "c3": col3})

    # Compute expected results by using the same logic as the implementation:
    # convert to str (they already are strings), then use .split() and len()
    expected_words = 0
    expected_chars = 0
    for col in ["c1", "c2", "c3"]:
        series = df[col].astype(str)  # mirrors function behavior
        # sum of character counts
        expected_chars += int(series.str.len().sum())
        # sum of words per row
        expected_words += int(series.str.split().str.len().sum())

    # Use the function and assert results match the expected computation
    words, chars = calculate_text_metrics(df, ["c1", "c2", "c3", "nonexistent_col"])
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import pandas as pd  # real class constructor for DataFrame
# imports
import pytest  # used for our unit tests
from langflow.api.utils.kb_helpers import calculate_text_metrics


def test_basic_single_column_counts_words_and_characters():
    # Create a simple DataFrame with one text column using the real pandas DataFrame constructor
    df = pd.DataFrame({"text": ["hello world", "foo"]})
    # Call the function under test
    words, chars = calculate_text_metrics(df, ["text"])


def test_multiple_columns_and_ignores_missing_columns():
    # DataFrame with two text columns and one numeric column
    df = pd.DataFrame(
        {
            "a": ["one two", "three"],
            "b": ["x y z", ""],
            "num": [1, 2],  # numeric column should not be counted unless listed
        }
    )
    # Include an extra non-existent column name 'missing' which should be skipped silently
    words, chars = calculate_text_metrics(df, ["a", "b", "missing"])


def test_empty_dataframe_and_empty_text_columns_returns_zeros():
    # Empty DataFrame with no columns
    df_empty = pd.DataFrame()
    # When text_columns is empty
    w1, c1 = calculate_text_metrics(df_empty, [])
    # When text_columns contains names not present in the DataFrame: they should be ignored
    w2, c2 = calculate_text_metrics(df_empty, ["nonexistent", "also_missing"])


def test_none_and_nan_and_numeric_values_are_cast_to_strings_and_counted():
    # The implementation calls astype(str) first, so None -> 'None' and nan -> 'nan'
    df = pd.DataFrame({"t": [None, float("nan"), 123]})
    words, chars = calculate_text_metrics(df, ["t"])


def test_whitespace_and_special_characters_counted_correctly():
    # Strings with newlines, tabs, and multiple spaces
    df = pd.DataFrame(
        {
            "s": [
                "a\nb\tc  d",  # a newline and a tab and double space -> 4 words
                "\tleading and trailing \n",  # words: "leading","and","trailing" -> 3 words
            ]
        }
    )
    words, chars = calculate_text_metrics(df, ["s"])
    # Count words explicitly to make the assertion clear and robust
    expected_words = sum(len(str(val).split()) for val in df["s"])
    expected_chars = sum(len(str(val)) for val in df["s"])


def test_large_scale_1000_rows_multiple_columns():
    # Construct 1000 rows to validate scaling up to the requested size
    n_rows = 1000
    # Column a: 5 occurrences of 'word ' (with trailing space); split() will count 5 words
    # Column b: single token of 10 'x' characters -> 1 word per row
    col_a = [("word " * 5) for _ in range(n_rows)]  # each entry length = 5 * len("word ") = 25 chars
    col_b = [("x" * 10) for _ in range(n_rows)]  # each entry length = 10 chars
    df = pd.DataFrame({"a": col_a, "b": col_b})
    words, chars = calculate_text_metrics(df, ["a", "b"])


def test_order_of_text_columns_does_not_change_result():
    # Ensure different ordering of columns in the list yields the same totals
    df = pd.DataFrame({"c1": ["one two"], "c2": ["three four five"]})
    codeflash_output = calculate_text_metrics(df, ["c1", "c2"]); res1 = codeflash_output
    codeflash_output = calculate_text_metrics(df, ["c2", "c1"]); res2 = codeflash_output


def test_mixed_type_columns_handled_and_counted_as_strings():
    df = pd.DataFrame({"m": [100, 200.5, "text"]})
    words, chars = calculate_text_metrics(df, ["m"])
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.

import pandas as pd  # used to construct DataFrame inputs
# imports
import pytest  # used for our unit tests
from langflow.api.utils.kb_helpers import KBAnalysisHelper


def test_schema_prioritized_columns_present():
    # Create a DataFrame with several columns including 'a' which the schema will mark as text
    df = pd.DataFrame({"a": ["hello"], "b": [1], "text": ["x"]})
    # Schema marks 'a' as vectorize=True and data_type='string'; also include a missing column
    schema = [
        {"column_name": "a", "vectorize": True, "data_type": "string"},
        {"column_name": "missing", "vectorize": True, "data_type": "string"},
    ]
    # Expect the helper to return only 'a' because it's the only schema-declared text column present in df
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema); result = codeflash_output


def test_schema_ignored_when_not_vectorize_or_not_string_and_common_name_matches():
    # DataFrame contains a common-name column 'Content' (mixed-case)
    df = pd.DataFrame({"id": [1], "Content": ["hello world"], "other": [2]})
    # Schema exists but does not mark any valid text columns (vectorize=False or wrong data_type). Should fall back.
    schema = [
        {"column_name": "id", "vectorize": False, "data_type": "string"},
        {"column_name": "other", "vectorize": False, "data_type": "integer"},
    ]
    # With no effective schema text columns, the helper should detect 'Content' by case-insensitive common names
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema); result = codeflash_output


def test_common_name_detection_multiple_order_preserved():
    # DataFrame with multiple common-name columns in a particular order and mixed casing
    df = pd.DataFrame({"Text": ["a"], "alpha": [0], "CONTENT": ["b"], "Document": ["c"], "z": [1]})
    # No schema_data provided -> should match common names in the DataFrame column order
    codeflash_output = KBAnalysisHelper._get_text_columns(df); result = codeflash_output


def test_fallback_to_object_dtype_when_no_schema_and_no_common_names():
    # Create DataFrame with one object dtype column ('col1') and one numeric column ('col2')
    df = pd.DataFrame({"col1": ["s1", "s2"], "col2": [1, 2]})
    # No schema and no common-name columns -> should return columns whose dtype is object
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output


def test_empty_dataframe_returns_empty_list():
    # Empty DataFrame with no columns
    df = pd.DataFrame()
    # Neither schema nor columns -> should return empty list
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output


def test_schema_columns_not_in_df_falls_back_to_common_names():
    # Schema points to a column not present in df; df contains 'chunk' so should be detected via common names
    df = pd.DataFrame({"chunk": ["part1"], "other": [1]})
    schema = [{"column_name": "not_present", "vectorize": True, "data_type": "string"}]
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=schema); result = codeflash_output



def test_large_schema_search_performance_and_correctness():
    # Build a large DataFrame with 1000 columns; include a special target column 'col_500' that the schema will mark
    data = {f"col_{i}": [f"v{i}"] for i in range(1000)}
    df = pd.DataFrame(data)
    # Create a large schema: most entries are not vectorized, except one that should be selected
    schema = [
        {"column_name": f"col_{i}", "vectorize": False, "data_type": "string"} for i in range(1000)
    ]
    # Mark a single column as vectorize True and data_type 'string'
    schema[500] = {"column_name": "col_500", "vectorize": True, "data_type": "string"}
    # The helper should return only the single column marked for vectorization and present in df
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=schema); result = codeflash_output


def test_large_dataframe_object_dtype_returns_all_object_columns():
    # Construct a DataFrame with 1000 object-typed columns (strings)
    n = 1000
    data = {f"obj_{i}": [f"v{i}"] for i in range(n)}
    df = pd.DataFrame(data)
    # No schema and no common-name columns; should return all columns because all are object dtype
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.
#------------------------------------------------
import pandas as pd  # DataFrame construction and dtypes
# imports
import pytest  # used for our unit tests
from langflow.api.utils.kb_helpers import KBAnalysisHelper


def test_schema_data_prefers_vectorized_string_columns_basic():
    # Create a DataFrame with some columns present
    df = pd.DataFrame(
        {
            "a": [1],  # numeric column
            "text_col": ["hello"],  # object/string column
            "b": ["x"],  # object/string column
        }
    )
    # schema_data lists three columns; two exist in df, one does not.
    schema_data = [
        {"column_name": "b", "vectorize": True, "data_type": "string"},
        {"column_name": "a", "vectorize": True, "data_type": "string"},
        {"column_name": "missing", "vectorize": True, "data_type": "string"},
    ]
    # Expect the function to return the intersection of schema order with df.columns:
    # order preserved from schema list but only existing columns kept -> ['b','a']
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data); result = codeflash_output


def test_schema_data_ignores_non_vectorized_or_non_string_entries():
    # Construct a DataFrame with two columns
    df = pd.DataFrame({"c": ["one"], "text": ["two"]})
    # schema_data includes a valid entry for 'c' and two invalid entries
    schema_data = [
        {"column_name": "c", "vectorize": True, "data_type": "string"},
        {"column_name": "text", "vectorize": False, "data_type": "string"},  # vectorize False -> ignore
        {"column_name": "text", "vectorize": True, "data_type": "int"},  # not string -> ignore
    ]
    # Only 'c' should be selected from schema_data
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data); result = codeflash_output


def test_schema_data_with_entries_not_in_dataframe_returns_empty_list_no_fallback():
    # When schema_data yields text_columns but none are in df.columns, the function returns []
    df = pd.DataFrame({"Content": ["alpha"], "other": ["beta"]})
    # schema_data contains valid vectorize/string entries but none of these columns exist in df
    schema_data = [
        {"column_name": "not_here", "vectorize": True, "data_type": "string"},
        {"column_name": "also_missing", "vectorize": True, "data_type": "string"},
    ]
    # Because text_columns from schema_data is non-empty, the code will attempt to filter them by df.columns
    # and return the filtered list (which should be empty) instead of falling back to common names.
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data); result = codeflash_output


def test_common_names_detection_case_insensitive_and_preserve_df_order():
    # Create DataFrame with common names in mixed case and specific order
    df = pd.DataFrame(
        {
            "id": [1],
            "Document": ["doc1"],
            "Text": ["txt1"],
            "other": ["o"],
            "chunk": ["c"],  # also a common name (lowercase)
        }
    )
    # No schema_data provided, so common name detection should run.
    # It should be case-insensitive and preserve the order of df.columns.
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output


def test_fallback_to_object_dtype_only_includes_object_series():
    # Column 'b' has dtype object (default for Python strings), 'c' has pandas string dtype
    df = pd.DataFrame(
        {
            "a": [1, 2],  # int -> not object
            "b": ["x", "y"],  # object dtype (default)
            "c": pd.Series(["s1", "s2"], dtype="string"),  # pandas StringDtype, not 'object'
        }
    )
    # No schema_data and no common names -> final fallback should return only columns with dtype == 'object'
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output


def test_empty_dataframe_returns_empty_list():
    # Empty DataFrame with no columns
    df_empty = pd.DataFrame()
    # No schema_data -> should return empty list
    codeflash_output = KBAnalysisHelper._get_text_columns(df_empty, schema_data=None); result = codeflash_output

    # If schema_data is an empty list (falsy), behavior is same as None -> empty DataFrame still returns []
    codeflash_output = KBAnalysisHelper._get_text_columns(df_empty, schema_data=[]); result2 = codeflash_output


def test_schema_data_none_and_empty_list_behave_same_as_no_schema():
    # DataFrame with a common-name column to confirm fallback path works when schema_data is None or empty list
    df = pd.DataFrame({"content": ["a"], "other": ["b"]})
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); res_none = codeflash_output
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=[]); res_empty = codeflash_output


def test_non_string_object_dtype_values_still_count_as_object():
    # Column 'obj' will contain dictionaries; dtype should be 'object' and therefore be selected by fallback
    df = pd.DataFrame({"obj": [{"k": 1}, {"k": 2}], "num": [1, 2]})
    codeflash_output = KBAnalysisHelper._get_text_columns(df); res = codeflash_output


def test_large_scale_schema_and_dataframe_intersection_preserves_schema_order():
    # Build a large DataFrame with 1000 columns named col0..col999
    n = 1000
    data = {}
    for i in range(n):
        # Use small lists for values; dtype will be object because they are Python strings
        data[f"col{i}"] = [f"v{i}"]
    df = pd.DataFrame(data)

    # Build schema_data of length 1000 where every even-indexed column is vectorize True/string
    schema_data = []
    expected = []
    # Put some names not in df as well to ensure filtering works
    for i in range(n):
        entry = {
            "column_name": f"col{i}" if (i % 3 != 0) else f"missing_{i}",  # every 3rd is missing
            "vectorize": (i % 2 == 0),  # even indices vectorize True
            "data_type": "string" if (i % 5 != 0) else "int",  # some are not strings
        }
        schema_data.append(entry)
        # Determine expected: column must be vectorize True, data_type 'string', and exist in df
        if entry["vectorize"] and entry["data_type"] == "string" and entry["column_name"] in df.columns:
            expected.append(entry["column_name"])

    # Call the helper and verify it returns the expected intersection in schema order
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data); result = codeflash_output


def test_large_scale_common_names_detection_in_big_dataframe():
    # Build a large DataFrame with 1000 columns but include a few common-name columns scattered
    n = 1000
    cols = [f"col{i}" for i in range(n)]
    # Insert some common names at specific positions
    cols[10] = "Text"
    cols[200] = "content"
    cols[999] = "Document"
    # Create DataFrame with these columns
    df = pd.DataFrame({c: [f"value_{i}"] for i, c in enumerate(cols)})
    # No schema_data -> should detect the three common-name columns in df order
    codeflash_output = KBAnalysisHelper._get_text_columns(df, schema_data=None); result = codeflash_output
# codeflash_output is used to check that the output of the original code is the same as that of the optimized code.